﻿using Confluent.Kafka;
using Md.Demo.Entity.MongoModels;
using Md.Demo.Tools;
using Microsoft.Extensions.Hosting;
using NLog;
using System;
using System.Threading;
using System.Threading.Tasks;

namespace Md.Demo.Service.BackJob
{

    public class KafkaTestBackJob : BackgroundService
    {
        private readonly Logger _logger = LogManager.GetLogger("logger");
        private readonly MongoContext _mongoContext;
        public KafkaTestBackJob(MongoContext mongoContext)
        {
            _mongoContext = mongoContext;
        }
        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            Task.Run(() =>
            {
                const string topicName = "test";
                var conf = new ConsumerConfig
                {
                    GroupId = "test-group",
                    BootstrapServers = "47.96.31.8:9092",
                    AutoOffsetReset = AutoOffsetReset.Earliest,
                    EnableAutoOffsetStore = false//<----this
                };

                using var c = new ConsumerBuilder<Ignore, string>(conf).Build();
                c.Subscribe(topicName);
                try
                {
                    while (true)
                    {
                        try
                        {
                            var consumeResult = c.Consume(stoppingToken);
                            _logger.Info($"Consumed message '{consumeResult.Message.Value}' at: '{consumeResult.TopicPartitionOffset}'.");
                            var message = consumeResult.Message.Value;
                            var job = message.ToObject<Job>();
                            CreateJob(job);
                            c.StoreOffset(consumeResult);
                        }
                        catch (ConsumeException e)
                        {
                            _logger.Error($"Error occured: {e.Error.Reason}");
                        }
                    }

                }
                catch (OperationCanceledException)
                {
                    c.Close();
                }
            });

            return Task.CompletedTask;
        }

        private Task CreateJob(Job request)
        {
            var jobCol = _mongoContext.Collection<Job>();
            jobCol.UpsertOne(request);
            return Task.CompletedTask;
        }
    }

}
